package com.lagou.sale.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lagou.sale.constant.Constant;
import com.lagou.sale.dao.JedisDao;
import com.lagou.sale.dao.OrderDao;
import com.lagou.sale.pojo.Order;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.List;

@Service
public class OrderConsumer {

    @Resource(name = "norCon")
    private DefaultMQPushConsumer normalConsumer;
    @Resource(name = "timeoutCon")
    private DefaultMQPushConsumer timeoutConsumer;
    @Autowired
    private DefaultMQProducer producer;
    @Autowired
    private JedisDao jedisDao;
    @Autowired
    private OrderDao orderDao;
    private ObjectMapper objectMapper = new ObjectMapper();

    @PostConstruct
    public void init() throws MQClientException {
        normalConsumer.setMessageListener((MessageListenerConcurrently)this::consumerOrderNormally);
        timeoutConsumer.setMessageListener((MessageListenerConcurrently)this::consumerOrderTimeout);
        normalConsumer.start();
        timeoutConsumer.start();
    }

    public ConsumeConcurrentlyStatus consumerOrderNormally(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list) {
            // 恢复订单类
            byte[] body = msg.getBody();
            Order order;
            try {
                order = objectMapper.readValue(body, Order.class);
            } catch (IOException e) {
                e.printStackTrace();
                continue;
            }

            // 处理订单请求
            order.setCreateTime(new Date());
            int num = orderDao.createOrder(order);
            if (num == 0) {
                System.out.println("order: "+order.getOrderId()+" already exist");
            } else {
                pushTimeoutMsgToMq(order);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public ConsumeConcurrentlyStatus consumerOrderTimeout(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list) {
            // 恢复订单类
            byte[] body = msg.getBody();
            Order order;
            try {
                order = objectMapper.readValue(body, Order.class);
            } catch (IOException e) {
                e.printStackTrace();
                continue;
            }

            // 将未支付状态的订单设置到已废弃（其他情况视为已经处理过）
            int num = orderDao.dropOrder(order.getOrderId());
            if (num == 1) {
                System.out.println("order: "+order.getOrderId()+" drop successfully");
                // 恢复库存
                jedisDao.addStock(order.getItemNum());
            } else {
                System.out.println("order: "+order.getOrderId()+" status is changed(paid or drop or not exist)");
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    private void pushTimeoutMsgToMq(Order order) {
        // 提交超时订单
        byte[] bytes = new byte[0];
        try {
            bytes = objectMapper.writeValueAsBytes(order);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

        Message message = new Message(Constant.OrderTimeoutTopic, bytes);
        message.setDelayTimeLevel(2);
        try {
            producer.send(message);
        } catch (MQClientException | InterruptedException | MQBrokerException | RemotingException e) {
            e.printStackTrace();
        }
        System.out.println("send timeout msg success, order: "+order.getOrderId());
    }
}
